package com.deepglint.transform;

import com.deepglint.beans.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.logging.Logger;

/**
 * @author mj
 * @version 1.0
 * @date 2021-11-15 21:53
 * 多流转换算子
 */
public class TransformTest_MultipleStreams {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String inputPath = "C:\\Users\\马军\\Desktop\\Idea-workspace\\flink\\src\\main\\resources\\source.txt";
        DataStream<String> inputStream = env.readTextFile(inputPath);

        // 转换成SensorReading类型
        DataStream<SensorReading> streamMap = inputStream.map(value -> {
            String[] split = value.split(",");
            return new SensorReading(split[0], split[1], new Long(split[2]), new Double(split[3]));
        });

        System.out.println("【-----分流开始-----】");
        //1. 分流操作，按照30度为分界值，分流
        // 通过OutputTag和context来新增一条打标签的流，可以通过getSideOutput和OutputTag获取打标签的流
        // collect相当于filter，对原来的流做一个过滤
        // ctx的返回值，是根据outputTag类型决定的
        final OutputTag<SensorReading> outputTagHigh = new OutputTag<SensorReading>("high") {
        };
        final OutputTag<SensorReading> outputTagLower = new OutputTag<SensorReading>("lower") {
        };
        SingleOutputStreamOperator<Object> other_stream = streamMap.process(new ProcessFunction<SensorReading, Object>() {
            @Override
            public void processElement(SensorReading value, ProcessFunction<SensorReading, Object>.Context ctx, Collector<Object> out) throws Exception {
                if (value.getTemperature() < 100) {
                    out.collect(value);
                }
                if (value.getTemperature() > 30) {
                    ctx.output(outputTagHigh, value);
                } else {
                    ctx.output(outputTagLower, value);
                }
            }
        });

        DataStream<SensorReading> lowerStream = other_stream.getSideOutput(outputTagLower);
        DataStream<SensorReading> highStream = other_stream.getSideOutput(outputTagHigh);

        lowerStream.print();
        highStream.print();
        other_stream.print();

        System.out.println("【-----分流结束-----】");


        System.out.println("【-----connect合流开始-----】");
        //2. 合流，将高温流转换为二元组类型，于低温流连接合并之后，输出一个状态信心
        SingleOutputStreamOperator<Tuple2<String, Double>> warnStream = highStream.map(new MapFunction<SensorReading, Tuple2<String, Double>>() {
            @Override
            public Tuple2<String, Double> map(SensorReading value) throws Exception {
                return new Tuple2<>(value.getId(), value.getTemperature());
            }
        });

        // 2.合并后的流(一流中两种类型,只能连接两条流)
        ConnectedStreams<Tuple2<String, Double>, SensorReading> connectedStreams = warnStream.connect(lowerStream);

        SingleOutputStreamOperator<Object> resultMap = connectedStreams.map(new CoMapFunction<Tuple2<String, Double>, SensorReading, Object>() {
            @Override
            public Object map1(Tuple2<String, Double> value) throws Exception {
                return new Tuple3<>(value.f0, value.f1, "high temp warning");
            }

            @Override
            public Object map2(SensorReading value) throws Exception {
                return new Tuple2<>(value.getId(), "normal");
            }
        });

        resultMap.print();
        System.out.println("【-----connect合流结束-----】");

        System.out.println("【-----union合流开始-----】");

        // 3、union联合多条流
        DataStream<SensorReading> unionStream = highStream.union(lowerStream);

        unionStream.print();

        System.out.println("【-----union合流结束-----】");

        env.execute();
    }
}
